package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"net/http"
	"os"
	"sort"
	"strconv"
	"strings"
	"sync"
	"github.com/hashicorp/memberlist"
)

var (
	mtx        sync.RWMutex
	members    = flag.String("members", "", "comma seperated list of members")
	http_port       = flag.Int("http_port", 4001, "http port")
	gossip_port       = flag.Int("gossip_port", 8001, "gossip port")
	items      = map[string]string{}
	broadcasts *memberlist.TransmitLimitedQueue
	mlist *memberlist.Memberlist
)



func position()int{
	all := mlist.Members()
	sort.Slice(all, func(i, j int) bool {
		return all[i].Name < all[j].Name
	})

	k := 0
	for _, n := range all {
		if n.Name == mlist.LocalNode().Name {
			break
		}
		k++
	}
	return k
}

type userdata struct {
	msg    []byte
	notify chan<- struct{}
}

type delegate struct{}

type update struct {
	Action string // add, del
	Data   map[string]string
}

func init() {
	flag.Parse()
}

func (b *userdata) Invalidates(other memberlist.Broadcast) bool {
	fmt.Println(11111)
	return false
}

func (b *userdata) Message() []byte {
	fmt.Println(22222222)
	fmt.Println(b.msg)
	return b.msg
}

func (b *userdata) Finished() {
	fmt.Println(33333333)
	if b.notify != nil {
		close(b.notify)
	}
}

func (d *delegate) NodeMeta(limit int) []byte {		//limit默认为512
	fmt.Println(4444444444)
	return []byte{}
}

func (d *delegate) NotifyMsg(b []byte) {		//接收广播发出来的增量数据
	fmt.Println(55555555)
	if len(b) == 0 {
		return
	}

	switch b[0] {
	case 'd': // data
		var updates []*update
		if err := json.Unmarshal(b[1:], &updates); err != nil {
			return
		}
		fmt.Println(updates)
		mtx.Lock()
		for _, u := range updates {
			for k, v := range u.Data {
				fmt.Println(k, v)
				switch u.Action {
				case "add":
					items[k] = v
				case "del":
					delete(items, k)
				}
			}
		}
		mtx.Unlock()
	}
}

func (d *delegate) GetBroadcasts(overhead, limit int) [][]byte {
	//fmt.Println(666666666)
	return broadcasts.GetBroadcasts(overhead, limit)
}

func (d *delegate) LocalState(join bool) []byte {			//实时上报自身的全量数据
	fmt.Println(77777777)
	fmt.Println(position())
	mtx.RLock()
	m := items
	mtx.RUnlock()
	b, _ := json.Marshal(m)
	return b
}

func (d *delegate) MergeRemoteState(buf []byte, join bool) {		//实时获取其他节点的全量数据合并本地
	fmt.Println(88888888)
	if len(buf) == 0 {
		return
	}
	fmt.Println(join)
	if !join {
		return
	}
	var m map[string]string
	if err := json.Unmarshal(buf, &m); err != nil {
		return
	}
	fmt.Println(m)
	mtx.Lock()
	for k, v := range m {
		fmt.Println(k, v)
		items[k] = v
	}
	mtx.Unlock()
}

func addHandler(w http.ResponseWriter, r *http.Request) {
	r.ParseForm()
	key := r.Form.Get("key")
	val := r.Form.Get("val")
	mtx.Lock()
	items[key] = val
	mtx.Unlock()

	b, err := json.Marshal([]*update{
		&update{
			Action: "add",
			Data: map[string]string{
				key: val,
			},
		},
	})

	if err != nil {
		http.Error(w, err.Error(), 500)
		return
	}
	//广播数据
	broadcasts.QueueBroadcast(&userdata{
		msg:    append([]byte("d"), b...),
		notify: nil,
	})
}

func delHandler(w http.ResponseWriter, r *http.Request) {
	r.ParseForm()
	key := r.Form.Get("key")
	mtx.Lock()
	delete(items, key)
	mtx.Unlock()

	b, err := json.Marshal([]*update{
		&update{
			Action: "del",
			Data: map[string]string{
				key: "",
			},
		},
	})

	if err != nil {
		http.Error(w, err.Error(), 500)
		return
	}

	broadcasts.QueueBroadcast(&userdata{
		msg:    append([]byte("d"), b...),
		notify: nil,
	})
}

func getHandler(w http.ResponseWriter, r *http.Request) {
	r.ParseForm()
	key := r.Form.Get("key")
	mtx.RLock()
	val := items[key]
	mtx.RUnlock()
	w.Write([]byte(val))
}

func start() error {
	hostname, _ := os.Hostname()
	c := memberlist.DefaultLocalConfig()
	c.Delegate = &delegate{}
	c.BindPort = *gossip_port
	c.Name = hostname + "-" + strconv.Itoa(*gossip_port)
	//创建gossip网络
	m, err := memberlist.Create(c)
	if err != nil {
		return err
	}
	mlist = m
	//第一个节点没有member，但从第二个开始就有member了
	if len(*members) > 0 {
		parts := strings.Split(*members, ",")
		_, err := m.Join(parts)
		if err != nil {
			return err
		}
	}
	broadcasts = &memberlist.TransmitLimitedQueue{
		NumNodes: func() int {
			return m.NumMembers()
		},
		RetransmitMult: 3,
	}
	node := m.LocalNode()
	fmt.Printf("Local member %s:%d\n", node.Addr, node.Port)
	return nil
}

func main() {
	if err := start(); err != nil {
		fmt.Println(err)
	}
	http.HandleFunc("/add", addHandler)
	http.HandleFunc("/del", delHandler)
	http.HandleFunc("/get", getHandler)
	fmt.Printf("Listening on :%d\n", *http_port)
	if err := http.ListenAndServe(fmt.Sprintf(":%d", *http_port), nil); err != nil {
		fmt.Println(err)
	}
}


//在一个终端执行go run gossip.go --http_port=4001 --gossip_port=8001 --members=10.20.0.10:8001,10.20.0.10:8002
//另一个终端执行go run gossip.go --http_port=4002 --gossip_port=8002 --members=10.20.0.10:8001,10.20.0.10:8002

//测试：
//curl "http://localhost:4001/add?key=foo&val=bar"
//停掉第一个
//curl "http://localhost:4002/add?key=foo&val=bar2"
//启动第一个
//curl "http://localhost:4001/get?key=foo"